package cn.wx.read.kafka.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.csvreader.CsvReader;
import cn.hutool.core.lang.Singleton;
import cn.wx.common.utils.TypeConversion;
import cn.wx.read.kafka.pool.ExecutorPool;
import cn.wx.read.kafka.pool.Gone;
import cn.wx.read.neo4j.entity.Relation;
import cn.wx.read.neo4j.service.Neo4jImportDataService;

/**
 * 建立专家和专利的关系
 * 写入neo4j
 */
@Service
public class RelationEfoService {
	
	Logger logger = LoggerFactory.getLogger(RelationEfoService.class);
	
	@Autowired
	Neo4jImportDataService neo4jImportDataService;

	ExecutorPool executorPool = Singleton.get(ExecutorPool.class);
	
	private final static String zoneName_efo = "ZoneRelationEfo_";
	private final static String zoneName_efo_paperCount = "ZoneRelationEfoPaperCount_";
	private final static String zoneName_efo_rank = "ZoneRelationEfoRank_";
	
	/**
	 * 开始分片
	 * filePath 文件所在路径
	 * fileName 文件名称
	 */
	public void consumeZone(String filePath, String fileName, String dataSource) {
		
		/**
		 * 获取分片的个数
		 */
		Integer size = executorPool.zoneCsv(filePath, fileName, dataSource);
		
		executorPool.initPool();
		
		/**
         * 分片完毕，启动线程
         */
		int sz = size / ExecutorPool.tc;
		
		if(sz == 0) {
			for(int i=0;i<size;i++){
				executorPool.execute(new ThreadMultiReader(filePath,i,i,dataSource));
			}
		}else{
			int e=0;
			for(int i=0;i<size;i++){
			    if(e==sz) {
			    	executorPool.execute(new ThreadMultiReader(filePath,(i-e),i,dataSource));
			    	e=0;
			    }else{
			    	e++;
			    }
			}
			if(e>0){
				int a=size - e + 1;
				int b=size - 1;
				executorPool.execute(new ThreadMultiReader(filePath,(a>b?b:a),b,dataSource));
			}
		}

        executorPool.poolShutdown();
	}
	
	/**
	 * 内部类线程
	 */
	class ThreadMultiReader implements Runnable {

		private Integer s_;
		
		private Integer e_;
		
		private String fp_;
		
		private String dataSource_;
		
		public ThreadMultiReader(String fp_,Integer s_,Integer e_,String dataSource_) {
			this.s_ = s_;
			this.e_ = e_;
			this.fp_ = fp_;
			this.dataSource_ = dataSource_;
		}
		
		@Override
		public void run() {
			
			/**
			 * 开始处理
			 */
			for(int i=s_;i<=e_;i++) {
				
				switch (dataSource_) {
				
				case zoneName_efo:
					consumeRelationEfo(fp_ + zoneName_efo + i +".csv");
					break;
				case zoneName_efo_paperCount:
					consumeRelationEfoPaperCount(fp_ + zoneName_efo_paperCount + i +".csv");
					break;
				case zoneName_efo_rank:
					consumeRelationEfoRank(fp_ + zoneName_efo_rank + i +".csv");
					break;
					
				default:
					break;
				}
				
				
			}
			
		}

		public Integer getS_() {
			return s_;
		}

		public void setS_(Integer s_) {
			this.s_ = s_;
		}

		public Integer getE_() {
			return e_;
		}

		public void setE_(Integer e_) {
			this.e_ = e_;
		}

		public String getFp_() {
			return fp_;
		}

		public void setFp_(String fp_) {
			this.fp_ = fp_;
		}

		public String getDataSource_() {
			return dataSource_;
		}

		public void setDataSource_(String dataSource_) {
			this.dataSource_ = dataSource_;
		}
		
	}
	
	/**
	 * 解析专家和领域的关系
	 * @param filePath
	 */
	public void consumeRelationEfo(String filePath){
		
		/**
		 * 定义ne4j对象
		 */
		Relation ne = null;
		List<Relation> nes = new ArrayList<>();
		
		try {
			
			CsvReader csvReader = executorPool.getCsvReader(filePath);
            csvReader.readHeaders();
            
            while (csvReader.readRecord()){
            	
            	String line = csvReader.getRawRecord();
            	
				if(StringUtils.isBlank(line)) {
					continue;
				}
				/**
				 * 从csv中获取数据
				 */
				String expertId = csvReader.get("expertId");
				String fosId = csvReader.get("fosId");

				ne = new Relation();
				ne.setLeftId(expertId);
				ne.setRightId(fosId);
				nes.add(ne);
				
				/**
				 * 批量插入
				 * 2000一组
				 */
				if(nes.size() == Gone.Neo4j.getIc()) {
					try {
						Long st = System.currentTimeMillis();
						neo4jImportDataService.importBetchRelationEfo(nes);
						Long et = System.currentTimeMillis();
						logger.info("一组专家和领域关系耗时：" + (float)(et-st)/1000);
					}catch (Exception e) {
						/**
						 * 执行错误 插入日志
						 */
					}finally {
						nes.clear();
					}
				}
            }
            
            if(nes.size() > 0){
            	neo4jImportDataService.importBetchRelationEfo(nes);
            }
            /**
			 * 关闭读入流
			 */
			csvReader.close();
			csvReader = null;
			ne = null;
			nes = null;
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 解析专家和领域的关系paperCount
	 * @param filePath
	 */
	public void consumeRelationEfoPaperCount(String filePath){
		
		/**
		 * 定义ne4j对象
		 */
		Relation ne = null;
		List<Relation> nes = new ArrayList<>();
		
		try {
			
			CsvReader csvReader = executorPool.getCsvReader(filePath);
            csvReader.readHeaders();
            
            while (csvReader.readRecord()){
            	
            	String line = csvReader.getRawRecord();
            	
				if(StringUtils.isBlank(line)) {
					continue;
				}
				/**
				 * 从csv中获取数据
				 */
				String expertId = csvReader.get("expertId");
				String fosId = csvReader.get("fosId");
				Integer paperCount = TypeConversion.converInteger(csvReader.get("paperCount"));
				ne = new Relation();
				ne.setLeftId(expertId);
				ne.setRightId(fosId);
				ne.setPaperCount(paperCount);
				nes.add(ne);
				
				/**
				 * 批量插入
				 * 2000一组
				 */
				if(nes.size() == Gone.Neo4j.getIc()) {
					try {
						Long st = System.currentTimeMillis();
						neo4jImportDataService.importBetchRelationEfoPaperCount(nes);
						Long et = System.currentTimeMillis();
						logger.info("一组专家和领域关系上论文数paperCount耗时：" + (float)(et-st)/1000);
					}catch (Exception e) {
						/**
						 * 执行错误 插入日志
						 */
					}finally {
						nes.clear();
					}
				}
            }
            
            if(nes.size() > 0){
            	neo4jImportDataService.importBetchRelationEfoPaperCount(nes);
            }
            /**
			 * 关闭读入流
			 */
			csvReader.close();
			csvReader = null;
			ne = null;
			nes = null;
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 解析专家和领域的关系rank
	 * @param filePath
	 */
	public void consumeRelationEfoRank(String filePath){
		
		/**
		 * 定义ne4j对象
		 */
		Relation ne = null;
		List<Relation> nes = new ArrayList<>();
		
		try {
			
			CsvReader csvReader = executorPool.getCsvReader(filePath);
            csvReader.readHeaders();
            
            while (csvReader.readRecord()){
            	
            	String line = csvReader.getRawRecord();
            	
				if(StringUtils.isBlank(line)) {
					continue;
				}
				/**
				 * 从csv中获取数据
				 */
				String expertId = csvReader.get("expertId");
				String fosId = csvReader.get("fosId");
				Integer rank = TypeConversion.converInteger(csvReader.get("rank"));
				ne = new Relation();
				ne.setLeftId(expertId);
				ne.setRightId(fosId);
				ne.setRank(rank);
				nes.add(ne);
				
				/**
				 * 批量插入
				 * 2000一组
				 */
				if(nes.size() == Gone.Neo4j.getIc()) {
					try {
						Long st = System.currentTimeMillis();
						neo4jImportDataService.importBetchRelationEfoRank(nes);
						Long et = System.currentTimeMillis();
						logger.info("一组专家和领域关系上论文数paperCount耗时：" + (float)(et-st)/1000);
					}catch (Exception e) {
						/**
						 * 执行错误 插入日志
						 */
					}finally {
						nes.clear();
					}
				}
            }
            
            if(nes.size() > 0){
            	neo4jImportDataService.importBetchRelationEfoRank(nes);
            }
            /**
			 * 关闭读入流
			 */
			csvReader.close();
			csvReader = null;
			ne = null;
			nes = null;
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
}
